package com.lili.collection.mq.config;

import com.lili.collection.core.config.InputConfig;
import com.lili.collection.core.domain.CollectCommonConfig;
import com.lili.collection.core.plugins.InputPlugin;
import com.lili.collection.mq.plugins.PulsarmqInputPlugin;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;

/**
 *描述: Pulsarmq输入插件配置
 *@date: 2023/9/26  9:21
 *@author: lili
 */
@Slf4j
@Data
public class PulsarmqInputConfig extends CollectCommonConfig implements InputConfig {

    /**
     * 服务地址
     */
    private String serviceUrl;

    /**
     * 命名空间角色token
     */
    private String token;

    /**
     * 主题
     */
    private String topic;

    /**
     * 消费组
     */
    private String consumerGroup;

    @Override
    public InputPlugin getPlugin() {
        return new PulsarmqInputPlugin();
    }

    @Override
    public void validate() {
        if(StringUtils.isBlank(serviceUrl)){
            throw new IllegalArgumentException("pulsar服务地址不能为空！");
        }
        if(StringUtils.isBlank(topic)){
            throw new IllegalArgumentException("pulsar主题不能为空！");
        }
        if(StringUtils.isBlank(consumerGroup)){
            throw new IllegalArgumentException("pulsar消费组不能为空！");
        }
        if(MapUtils.isEmpty(getAliasMap())){
            throw new IllegalArgumentException("pulsar别名映射不能为空！");
        }
        if(this.isIncrementalMode() && StringUtils.isBlank(this.getOffsetColumn())){
            throw new IllegalArgumentException("pulsar自增模式偏移列不能为空！");
        }
    }
}